package com.shujia.tf

import com.shujia.bean._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

class KcCarWindowReduce extends ProcessWindowFunction[CarWide, KcCarReduce, String, TimeWindow] {

  override def process(key: String, context: Context, elements: Iterable[CarWide], out: Collector[KcCarReduce]): Unit = {


    //窗口结束时间
    val ts = context.window.getEnd
    //统计车流量
    var count = 0L
    //总速度
    var sumSpeed = 0.0
    for (elem <- elements) {
      count += 1

      sumSpeed += elem.getSpeed
    }

    val kcCarReduce = new KcCarReduce

    kcCarReduce.setKcId(key)
    kcCarReduce.setTs(ts)
    kcCarReduce.setCount(count)
    kcCarReduce.setAvgSpeed(sumSpeed / count)

    //返回结果
    out.collect(kcCarReduce)

  }
}
